# -*-coding:utf-8-*-
import asyncio
import functools
import time
import threading
import typing
import typing

def thread_sleep(t):
    time.sleep(t)

async def task(task_index):
    if que.empty():
        return 'break_task_%d-队列已空' % task_index
    item = await que.get()
    print('exec_task_%d-获取-%d-当前线程-%s' % (task_index, item, threading.current_thread().name))
    await asyncio.sleep(1)  # 模拟耗时时间
    """
    可以将一个同步函数变为异步,利用functools.partial参入函数名和参数生成一个不带参数的函数
    如:run_in_executor(None, functools.partial(requests.get, url="", ...))
    但此方式内部是创建了线程池,不推荐使用,异步网络请求用aiohttp库
    """
    future = asyncio.get_running_loop().run_in_executor(None, functools.partial(thread_sleep, 1))
    await future
    print('finish_task_%d-获取-%d-当前线程-%s' % (task_index, item, threading.current_thread().name))
    return 'complete_task_%d' % task_index


que = asyncio.Queue()


# 按照顺序异步执行,全部协程执行完再返回结果
async def main():
    for index in range(10):
        await que.put(index)
    tasks = [task(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(results)


# as_completed能按照协程的完成顺序返回
async def main2():
    for index in range(10):
        await que.put(index)
    tasks = [task(i) for i in range(5)]
    for t in asyncio.as_completed(tasks):
        result = await t
        print(result)


# 调用asyncio.create_task()方法将协程包装成Task(Future子类)
async def main3():
    for index in range(10):
        await que.put(index)
    tasks = [asyncio.create_task(task(i)) for i in range(5)]
    # 不需要结果时可以用asyncio.wait等待任务执行结束
    # await asyncio.wait(tasks)
    for t in tasks:
        r = await t
        print(r)

# 通过任务组可以很方便的等待分组中所有任务完成
async def main4():
    for index in range(10):
        await que.put(index)
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(task(i)) for i in range(2)]
    print(f"tasks have completed: {tasks[0].result()}, {tasks[1].result()}")

# ------------在主线程中执行协程任务-------------
def run_sync():
    """
    asyncio.run和asyncio.get_event_loop().run_until_complete是在当前线程中执行协程的,会卡住当前线程
    """
    # 启动协程方式一
    asyncio.run(main4())

    # 启动协程方式二
    # asyncio.get_event_loop().run_until_complete(main())

    # 启动协程方式三，此方式可以拿到main()对应的future以中途取消任务
    # future = asyncio.get_event_loop().create_task(main())
    # asyncio.get_event_loop().run_until_complete(future)


# ------------在子线程中执行协程任务-------------
class AsyncioThread1(threading.Thread):
    def __init__(self):
        super().__init__()
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        # asyncio.Task为asyncio.Future子类
        self.future: asyncio.Task = None
    
    async def worker(self):
        print('start worker in %s' % threading.current_thread().name)
        await asyncio.sleep(2)
        print('stop worker')
    
    def stop(self):
        # 可以通过future.cancel()中途取消协程任务，可以用于实现下载任务暂停等功能
        cancelled = self.future.cancel()
        print('task_cancelled:', cancelled)
    
    def run(self):
        self.future = self.loop.create_task(self.worker())
        # 取消根层任务时run_until_complete不能正常等到task完成会触发asyncio.CancelledError，因此需要捕获异常
        try:
            self.loop.run_until_complete(self.future)
        except asyncio.CancelledError as e:
            print('except asyncio.CancelledError')
        finally:
            self.loop.run_until_complete(self.loop.shutdown_asyncgens())
            self.loop.close()
            print('event_loop closed:', self.loop.is_closed())


def run_async1():
    """
    直接在子线程中执行协程任务, 协程任务执行完成后子线程也将结束
    此种方式执行每执行一次协程任务就需要创建一个AsyncioThread1线程
    """
    asyncio_thread = AsyncioThread1()
    asyncio_thread.start()
    time.sleep(1)
    asyncio_thread.stop()
    time.sleep(4)  # 加个延时，防止协程还没执行完程序就结束了看不到输出结果


if __name__ == '__main__':
    run_sync()
    #run_async1()